package com.dyna.cp.binlog.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.dyna.cp.binlog.BinLogService;
import com.dyna.cp.binlog.BinaryLogClientRunner;
import com.dyna.cp.binlog.Event;
import com.dyna.cp.binlog.entity.BinLogEntity;
import com.dyna.cp.binlog.mapper.BinLogHistoryMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.io.Serializable;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

@Slf4j
@Service
public class BinLogServiceImpl implements BinLogService {

    private final BinLogHistoryMapper binLogHistoryMapper;

    @Autowired
    public BinLogServiceImpl(BinLogHistoryMapper binLogHistoryMapper) {
        this.binLogHistoryMapper = binLogHistoryMapper;
    }

    @Override
    public void handleWrite(List<Serializable[]> rows, BinaryLogClientRunner.TableData tableData, Map<String, Map<Integer, String>> metadata) {
        JSONObject jsonObject = writeOrDelete(rows, tableData, metadata);
        BinLogEntity saveEntity = new BinLogEntity(Event.WRITE.getKey(), tableData.getDatabase(), tableData.getTable(), null, JSON.toJSONString(jsonObject));
        binLogHistoryMapper.insert(saveEntity);
        log.info("binlog insert data: {}", JSON.toJSONString(jsonObject));
    }

    @Override
    public void handleUpdate(List<Map.Entry<Serializable[], Serializable[]>> rows, BinaryLogClientRunner.TableData tableData, Map<String, Map<Integer, String>> metadata) {
        Map<Integer, String> map = metadata.get(tableData.getDatabaseTable());
        if (CollectionUtils.isEmpty(map)) {
            return;
        }
        JSONObject dataObject = new JSONObject();
        JSONObject before = new JSONObject();
        for (Map.Entry<Serializable[], Serializable[]> row : rows) {
            List<Serializable> entriesBefore = Arrays.asList(row.getKey());
            List<Serializable> entriesAfter = Arrays.asList(row.getValue());
            // before
            JSONObject dataObjectBefore = getDataObject(entriesBefore, map);
            // after
            dataObject = getDataObject(entriesAfter, map);
            before.put("before", dataObjectBefore);
        }
        BinLogEntity updateBinlogEntity = new BinLogEntity(Event.UPDATE.getKey(), tableData.getDatabase(), tableData.getTable(), JSON.toJSONString(before.get("before")), JSON.toJSONString(dataObject));
        log.info("binlog insert data: {}", JSON.toJSONString(dataObject));
        binLogHistoryMapper.insert(updateBinlogEntity);
    }

    private JSONObject getDataObject(List<Serializable> message, Map<Integer, String> metadata) {
        JSONObject resultObject = new JSONObject();
        for (int i = 0; i < message.size(); i++) {
            Object value = null;
            Object serializable = message.get(i);
            String key = metadata.get(i);
            if (null != serializable) {
                Class<?> aClass = serializable.getClass();
                if (aClass.getName().equals("[B")) {
                    value = new String((byte[]) serializable);
                } else {
                    value = serializable;
                }

                if (key.equals("create_time") || key.equals("update_time")) {
                    value = (long) value - 8 * 60 * 60 * 1000;
                }

                resultObject.put(key, value);
            }
        }
        return resultObject;
    }

    @Override
    public void handleDelete(List<Serializable[]> rows, BinaryLogClientRunner.TableData tableData, Map<String, Map<Integer, String>> metadata) {
        JSONObject jsonObject = writeOrDelete(rows, tableData, metadata);
        log.info("binlog delete data: {}", JSON.toJSONString(jsonObject));
        BinLogEntity deleteBinlogEntity = new BinLogEntity(Event.DELETE.getKey(), tableData.getDatabase(), tableData.getTable(), null, JSON.toJSONString(jsonObject));
        binLogHistoryMapper.insert(deleteBinlogEntity);
    }

    private static JSONObject writeOrDelete(List<Serializable[]> rows, BinaryLogClientRunner.TableData tableData, Map<String, Map<Integer, String>> metadata) {
        Map<Integer, String> map = metadata.get(tableData.getDatabaseTable());
        if (CollectionUtils.isEmpty(map)) {
            return null;
        }
        JSONObject resultObject = new JSONObject();
        for (int i = 0; i < rows.size(); i++) {
            Serializable[] serializableList = rows.get(i);
            for (int j = 0; j < serializableList.length; j++) {
                Serializable serializable = serializableList[j];
                if (null != serializable) {
                    Object value = null;
                    Class<?> aClass = ((Object) serializable).getClass();
                    if (aClass.getName().equals("[B")) {
                        value = new String((byte[]) serializable);
                    } else {
                        value = serializable;
                    }
                    resultObject.put(map.get(j), value);
                }
            }
        }
        return resultObject;
    }

}
